package org.huang.rocketmq.demo.order;

import java.io.UnsupportedEncodingException;
import java.util.List;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.huang.rocketmq.demo.Constant;

/**
 * 这是一个发送顺序消息的rocketMQ提供者
 *   要发送顺序消息，我们要注意的是，send方法。要求传入MessageQueueSelector接口的实现类。
 *   这个实现类表示我们的消息要放入到topic的哪个队列中去。
 *
 *
 *   如下面的实例，则表示产生了5条消息，并且将这5条消息顺序的放到 topic=OrderTopicTest 的队列的下标为1的队列中去（默认情况下，一个topic会对应有4个队列）
 *
 *
 */
public class Producer {
    public static void main(String[] args) throws UnsupportedEncodingException {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("orderProducer");
            //指定连接的mq的nameserver地址
            producer.setNamesrvAddr(Constant.NAME_SERVER_ADDR);
            producer.start();

            for (int i = 1; i <=5; i++) {
                Message msg =
                        new Message("OrderTopicTest", "TAG-A", ("Order RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    /**
                     * 发送数据：如果要求顺序消费，将这些消息发送到同一个队列中去
                     * @param mqs  这个topic准备了多个消息队列（默认4个）
                     * @param msg  发送的消息
                     * @param arg  带过来的arg参数。
                     * @return
                     */
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        //选取其中的下标为1的队列，将消息放入到该队列中
                        return mqs.get(id);
                    }
                }, 1);//1是对标的下标

                System.out.printf("%s%n", sendResult);
            }

            producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
